package com.mtoolkit.socket.server;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;

import com.mtoolkit.socket.SocketRequest;

/**
 * 读线程。
 * 使用线程池技术，通过多个线程读取客户端数据，以充分利用网络数据传输的时间，提高读取效率。
 */
class NioServerReader extends Thread {
	
	private static int CAPACITY = 1024;
    private static List<SelectionKey> pool = new LinkedList<SelectionKey>();
    // event触发器
    private static NioServerEventNotifier notifier = NioServerEventNotifier.getNotifier();

    public NioServerReader() {
    }

    @Override
    public void run() {
        while (true) {
            SelectionKey key;
            try {
                synchronized (pool) {
                    while (pool.isEmpty()) {
                        pool.wait();
                    }
                    key = pool.remove(0);
                }

                // 读取数据
                read(key);
            } catch (InterruptedException ex) {
                continue;
            }
        }
    }

    /**
     * 读取数据。
     * @param key 选择键。
     */
    private void read(SelectionKey key) {
        // 读取客户端数据
        SocketChannel socketChannel = (SocketChannel) key.channel();

        try {
            byte[] clientData = readRequest(socketChannel);
            SocketRequest request = (SocketRequest) key.attachment();
//            request.setDateInput(clientData);
            request.setRequestData(clientData);
            notifier.fireOnReaded(request);
            // 提交给主控线程进行写处理
            NioSocketServer.processWriteRequest(key);
        } catch (Exception ex) {
            notifier.fireOnError(ex);
        }
    }

    /**
     * 读取客户端发送的请求中的数据。
     * @param socketChannel 客户端套接字通道。
     * @return 请求中的数据。
     * @throws java.io.IOException IO异常。
     */
    private byte[] readRequest(SocketChannel socketChannel) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(CAPACITY);

        int off = 0;
        int r = 0;
        byte[] datas = new byte[CAPACITY * 10];

        while (true) {
            buffer.clear();

            r = socketChannel.read(buffer);
            if (r == -1) {
                break;
            }

            if ((off + r) > datas.length) {
                datas = grow(datas, CAPACITY * 10);
            }

            byte[] buf = buffer.array();
            System.arraycopy(buf, 0, datas, off, r);

            off = off + r;
        }

        byte[] requestDatas = new byte[off];
        System.arraycopy(datas, 0, requestDatas, 0, off);

        return requestDatas;
    }

    /**
     * 数组扩容。
     * @param datas 源数组。
     * @param size 扩容的增量。
     * @return 扩容后的数组。
     */
    private byte[] grow(byte[] datas, int size) {
        byte[] temp = new byte[datas.length + size];
        System.arraycopy(datas, 0, temp, 0, datas.length);
        return temp;
    }

    /**
     * 处理客户请求，管理用户的连接池，并唤醒队列中的线程进行处理。
     * @param key 选择键。
     */
    public static void processRequest(SelectionKey key) {
        synchronized (pool) {
            pool.add(pool.size(), key);
            pool.notifyAll();
        }
    }
}
